/*
 *  proj3.c
 *  CS 505: Fall 2011
 *	PAXOS
 *  Ruchith Fernando and John Ross Wallrabenstein
 *
 */

#include "includes.h"

/*** Global Variables ***/
int port = 8080;
const int TIMEOUT_S_SEC = 10;
const int TIMEOUT_C_SEC = 10;
const int TIMEOUT_USEC = 2;
const int max_transaction_len = 10;
unsigned char* type; //Boolean Type in {l iff Coordinator,0 o.w.}
char* hostfile;
char* local_hostname;
char* coordinator_hostname;
int num_conn_hosts = 0;
extern int my_id;
const int host_length = 20;
fd_set master;
fd_set read_fds;
int fdmax;
int listener;
int newfd;
struct timeval tv;
int num_to_conn;
unsigned char* buffer;
extern int last_attempted;
extern int last_installed;
extern struct host_entry _hosts;
extern int host_count;
extern int curr_host_count;
extern int state;
int slow = 0;

/*** Function Prototypes ***/
void set_params(int arg_cnt, char* arg_vars[]);
void read_hosts();
void start_player();
char* get_hostname(const char* ipaddr);
void connect_to_all();
void exchange_messages();
void analyze_message(char* msg, char* author, int socket);
void send_to_all(char* msg);
int send_to_one(char* msg, char* host, int p);
void reclaim_memory();
void close_sockets();
int my_itoa(int val, char* buf);


/* Controls Program Flow */
int main(int argc, char* argv[]){

	/*Initialize data structures*/
	init_ds();

	/* Gather Parameters */
	set_params(argc,argv);

	/* Read Hosts File */
	read_hosts();
	
	/* Restore From Disk */
	restore();

	/* Start Player */
	start_player();

	/* Close Sockets */
	close_sockets();

	/* Reclaim Allocated Memory */
	reclaim_memory();
}


/* Deallocates Dynamic Memory */
void reclaim_memory(){

	free(hostfile);
	free(local_hostname);
	free(coordinator_hostname);
}

void start_player(){

	buffer = malloc(sizeof(char)*(max_transaction_len));

	connect_to_all();
	
	last_attempted++;
	shift_to_leader_election(last_attempted);
	
	exchange_messages();

}


/* Stores Parameters from Command Line */
void set_params(int arg_cnt, char* arg_vars[]){

	/* Validate Argument Length */
	//if(arg_cnt > 6 || arg_cnt < 5){

		//fprintf(stderr, "Usage: proj1 [-loyal, -traitor] -port PORT -hostfile HOSTFILE_PATH ORDER\n");
		//exit(1);
	//}
	
	if(arg_cnt > 6){
		DEBUGPRINT("SLOW MODE IS ON\n",NULL);
		slow = 1;
	}

	/* Get Local Hostname */
	FILE *cfp = popen("hostname", "r");
	local_hostname = malloc(host_length*sizeof(char)+1);
	if(cfp == NULL){
		DEBUGPRINT("Error: Unable to retrieve local machine hostname\n", NULL);
		exit(1);
	}else{
		while((fscanf(cfp,"%s\n",local_hostname)) != EOF){
			//DEBUGPRINT("Localhost: %s\n", local_hostname);
		}
		pclose(cfp);
	}

	//DEBUGPRINT( "\nPlayer Parameters:\n", NULL);

	/* Set Port */
	int tmp_port = atoi(arg_vars[2]);
	if(tmp_port < 1024 || tmp_port > 65536){
		DEBUGPRINT("Error: Port selected must be in range [1024,65536]\n", NULL);
		exit(1);
	}else {
		port = tmp_port;
		//DEBUGPRINT("Port: %d\n",port);
	}

	/* Store Hostfile Directory */
	hostfile = (char*) malloc(strlen(arg_vars[4])*sizeof(char)+1);
	strncpy(hostfile,arg_vars[4],strlen(arg_vars[4]));
	//DEBUGPRINT("Host Dir: %s\n",hostfile);

	type = (unsigned char*) malloc(sizeof(char)+1);

	type[0] = '0';
	type[1] = '\0';
}

/* Read and Store Hosts */
void read_hosts(){

	FILE *fp = fopen(hostfile, "r");

	if(fp == NULL){
		DEBUGPRINT( "Error: Could not open file %s\n", hostfile);
		exit(1);
	}

	coordinator_hostname = malloc(sizeof(char)*strlen(local_hostname+1));

	/* Read Hosts */
	while(!feof(fp)){
		char* line = malloc(sizeof(char)*(host_length+1));
		fscanf(fp,"%s\n",line);
		line[host_length] = '\0';

		if(host_count == 0){
			memcpy(coordinator_hostname,line,strlen(line));
			coordinator_hostname[host_length] = '\0';
		}
		
		/* Only Add OTHER (not local machine) Hosts */
		if(!strcmp(line,local_hostname)){
			my_id = host_count;
		}
		
		//DEBUGPRINT("Host: %s, Number: %d\n",line,host_count);
		
		add_host(line,-1);
	}

	if(0 == my_id){
		type[0] = '1';
		DEBUGPRINT("I am the Leader: %s\n",local_hostname);

		/* Set Timeout */
		tv.tv_sec = TIMEOUT_C_SEC;
		tv.tv_usec = TIMEOUT_USEC;

	}else{
		DEBUGPRINT("I am a Server (Non-Leader): %s\n",local_hostname);

		/* Set Timeout */
		tv.tv_sec = TIMEOUT_S_SEC;
		tv.tv_usec = TIMEOUT_USEC;
	}

	/* Store Connection Parameters */
	num_to_conn = host_count-1;

	fclose(fp);
	
	curr_host_count = host_count;
}

/* Given a string xxxxYY..(e.g. xinu01), return YY */
int get_hostnum(char* host){

	char chnum[3];
	chnum[0] = host[4];
	chnum[1] = host[5];
	chnum[2] = '\0';
	return atoi(chnum);
}

/* Form a Clique between Participants */
void connect_to_all(){

	DEBUGPRINT("\n*** Connecting to Participants ***\n", NULL);

	struct sockaddr_in remote;
	struct sockaddr_in remoteaddr;
	socklen_t addrlen;
	int nbytes, i;
	char remoteIP[15];

	listener = socket(AF_INET, SOCK_STREAM, 0);
	if(listener < 0){
		DEBUGPRINT( "Error: Could not open listener\n", NULL);
		exit(1);
	}

	bzero((char *) &remote, sizeof(remote));
	remote.sin_family = AF_INET;
	remote.sin_addr.s_addr = INADDR_ANY;
	remote.sin_port = htons(port);
	if (bind(listener, (struct sockaddr *) &remote,
			sizeof(remote)) < 0){
		DEBUGPRINT("Error: binding listener to socket\n", NULL);
		exit(1);
	}

	if(listen(listener, 5) < 0){
		DEBUGPRINT( "Error: Listener failed to bind\n", NULL);
		exit(1);
	}

	FD_SET(listener, &master);
	FD_SET(listener, &read_fds);

	fdmax = listener;

	/* Connect to Lower Host Numbers 
		(i.e. YY < XX, where xinuYY < (local)xinuXX)
	 */
	int j = 0;
	struct list_head *pos;
	struct host_entry *tmp;
	list_for_each(pos, &_hosts.list){

		tmp = list_entry(pos, struct host_entry, list);

		/* Skip Localhost and Commander */
		if(!strcmp(local_hostname,tmp->host_name))
			continue;

		if(tmp->id < my_id){

			//DEBUGPRINT("Connecting to %s\n",tmp->host_name);

			int newfd;
			struct sockaddr_in serv_addr;
			struct hostent *server;

			newfd = socket(AF_INET, SOCK_STREAM, 0);

			if(newfd < 0){
				DEBUGPRINT( "Error: Socket could not be opened\n", NULL);
				continue;
			}
			server = gethostbyname(tmp->host_name);
			if(server == NULL){
				DEBUGPRINT( "Error: Could not locate host %s\n", tmp->host_name);
				continue;
			}

			bzero((char*) &serv_addr, sizeof(serv_addr));
			serv_addr.sin_family = AF_INET;
			bcopy((char *)server->h_addr, 
					(char *)&serv_addr.sin_addr.s_addr,
					server->h_length);
			serv_addr.sin_port = htons(port);

			if (connect(newfd,(struct sockaddr *) &serv_addr,sizeof(serv_addr)) < 0){
				--j;
				continue;
			}

			//DEBUGPRINT("Connected to %s on %i\n",tmp->host_name,newfd);

			set_host_socket(tmp->host_name,newfd);
			num_to_conn--;

			/* Add Socket to Select's Watchlist */
			FD_SET(newfd, &master);
			if(newfd > fdmax){
				fdmax = newfd;
			}
		}
	}

	/* Listen for Connections from Higher Host Numbers 
		(i.e. YY > XX, where xinuYY > (local)xinuXX) */

	while(num_to_conn > 0){
	
		//DEBUGPRINT("Num To Contact: %d\n",num_to_conn);

		int num_ready = select(fdmax+1,&read_fds,NULL,NULL,&tv);

			addrlen = sizeof(remoteaddr);
			newfd = accept(listener,(struct sockaddr*) &remoteaddr, &addrlen);

			if(newfd < 0){
				DEBUGPRINT("Error: Didn't accept connection\n", NULL);
			}else{

				char str[host_length+1];
				const char* ipaddr = inet_ntop(AF_INET,&remoteaddr.sin_addr,str,sizeof(str));

				char* hname = get_hostname(ipaddr);

				FD_SET(newfd, &master);
				if(newfd > fdmax){
					fdmax = newfd;
				}

				//DEBUGPRINT("Connected to %s on %i\n", hname,newfd);
				
				set_host_socket(hname,newfd);
				num_to_conn--;
			}
	}
	
	DEBUGPRINT("Completely Connected\n", NULL);
}

/* Main Program Loop to Commit/Abort Transactions */
void exchange_messages(){

	DEBUGPRINT("\n*** Exchanging Messages ***\n", NULL);

	/* Receive/Send New Loop */
	for(;;){
		
		read_fds = master;

		if(!strcmp(type,"0")){
			tv.tv_sec = TIMEOUT_S_SEC;
			tv.tv_usec = TIMEOUT_USEC;
		}else{
			tv.tv_sec = TIMEOUT_C_SEC;
			tv.tv_usec = TIMEOUT_USEC;
		}

		/* See Which Sockets are Ready to Read */
		int num_ready = select(fdmax+1,&read_fds,NULL,NULL,&tv);
		
		if(num_ready==-1){
			DEBUGPRINT("Error: Select function failure\n", NULL);
			exit(1);
		}

		/* Loop Over Sockets */
		int i;
		for(i = listener; i <= fdmax; ++i){
		
			/* New Connections from Recovered Processes or Clients*/
			if(FD_ISSET(i,&read_fds) && (i == listener)){
				
				struct sockaddr_in remote;
				struct sockaddr_in remoteaddr;
				socklen_t addrlen;
				char remoteIP[15];
				
				addrlen = sizeof(remoteaddr);
				newfd = -1;
				newfd = accept(i,(struct sockaddr*) &remoteaddr, &addrlen);

				if(newfd < 0){
					DEBUGPRINT("Error: Didn't accept connection\n", NULL);
					DEBUGPRINT("Errno is: %d\n",errno);
				}else{
					
					char str[host_length+1];
					const char* ipaddr = inet_ntop(AF_INET,&remoteaddr.sin_addr,str,sizeof(str));

					char* hname = get_hostname(ipaddr);
					
					FD_SET(newfd, &master);
					FD_SET(newfd, &read_fds);
					if(newfd > fdmax){
						fdmax = newfd;
					}

					DEBUGPRINT("New Connection to %s on %i\n", hname,newfd);
					
					/* See How Many Bytes to Read */
					/*
					char* s = malloc(sizeof(char)*6);
					int b = read(newfd, s, 6);
					s[5] = '\0';

					int toread = atoi(s);
					
					DEBUGPRINT("Reading %d bytes from %s\n",toread,hname);

					/* Read the Message */
					/*
					bzero(buffer,sizeof(char)*max_transaction_len);
					int nbytes = read(newfd, buffer, toread);
					buffer[nbytes] = '\0';
					
					DEBUGPRINT("Read %s from %s\n",buffer,hname);
					
					analyze_message(buffer,hname,newfd);
					DEBUGPRINT("Returning after Fresh Connection\n",NULL);
					continue;*/
				}
			}

			//DEBUGPRINT("After New Connection\n",NULL);

			/* Check Sockets that are Ready */
			if((i != listener) && FD_ISSET(i, &read_fds)){
			
				//DEBUGPRINT("Checking Socket %d (Ready)\n",i);

				/* See How Many Bytes to Read */
				char* s = malloc(sizeof(char)*6);
				int b = read(i, s, 6);
				s[5] = '\0';

				int toread = atoi(s);

				/* Read the Message */
				bzero(buffer,sizeof(char)*max_transaction_len);
				int nbytes = read(i, buffer, toread);
				buffer[nbytes] = '\0';

				/* Find Sender Hostname */
				char* h = get_host_by_socket(i);

				/* Socket has Disconnected */
				if(nbytes == 0){
					DEBUGPRINT("Host %s Disconnected\n",h);
					remove_host(h);
					/* Remove Disconnected Socket from Select */
					FD_CLR(i,&master);
					FD_CLR(i,&read_fds);
						
					/* See if the Coordinator Died */
					if(!strcmp(coordinator_hostname,h)){
						DEBUGPRINT("Coordinator Has Died\n", NULL);
						shift_to_leader_election(last_attempted+1);
					}
				}else{
				
					/* Respond to Message Appropriately */
					analyze_message(buffer, h, i);
				}
			}
		}
	}
}

/* Responds to Received Messages Appropriately */
void analyze_message(char* msg, char* author, int socket){
	
	if(slow == 1){
		sleep(1);
	}
	
	int msg_type = get_message_type(msg);
	
	//DEBUGPRINT("Analyzing %s\n",msg);
	
	if(msg_type == VC){
	
		struct msg_view_change* tmp = deserialize_view_change(msg);
		
		// And Progress Timer Not Set? (Didn't Include This!!!)
		if(tmp->attempted > last_attempted){
			shift_to_leader_election(tmp->attempted);
			update_data_structures(msg);
		}else if(tmp->attempted == last_attempted){
		
			update_data_structures(msg);
			
			if(preinstall_ready(tmp->attempted) && tmp->attempted > last_installed){
			
				//Progress_Timer = Progress_Timer*2;
				//Set Progress Timer//
				
				int new_server = tmp->attempted % host_count;
				
				memcpy(coordinator_hostname,get_host_name(new_server),strlen(get_host_name(new_server)));
				coordinator_hostname[host_length] = '\0';
				
				//DEBUGPRINT("PreInstall Ready; New Leader is %s\n",coordinator_hostname);
				
				if(my_id == new_server){
					DEBUGPRINT("I AM THE ELECTED LEADER\n", NULL);
					shift_to_prepare_phase();
				}
			}
		}else{
			DEBUGPRINT("Attempted (%d) < Last Attempted (%d) -> Ignoring...\n",tmp->attempted,last_attempted);
		}
	
	}else if(msg_type == VC_PROOF){
	
		struct msg_vc_proof* tmp = deserialize_vc_proof(msg);
		
		if(tmp->installed > last_installed){
			last_attempted = tmp->installed;
			if(my_id == tmp->server_id){
				shift_to_prepare_phase();
			}else{
				shift_to_reg_non_leader();
			}
		}else{
			DEBUGPRINT("Installed (%d) <= Last Installed (%d) Ignoring...\n",tmp->installed,last_installed);
		}
	
	}else if(msg_type == PREP){
	
		struct msg_prepare* tmp = deserialize_prepare(msg);
		
		if(state == STATE_LEADER_ELECTION){
			//Install The View
			update_data_structures(msg);
			last_installed = tmp->view;
			struct msg_prepare_ok *pok = create_prepare_ok(my_id, last_installed);
			add_prep_ok(pok);
			shift_to_reg_non_leader();
			send_to_one(serialize_prepare_ok(pok),coordinator_hostname,1);
		}else{
			//Already Installed the View
			last_installed = tmp->view;
			struct msg_prepare_ok *pok = create_prepare_ok(my_id, last_installed);
			send_to_one(serialize_prepare_ok(pok), coordinator_hostname,1);
		}
	
	}else if(msg_type == PREP_OK){
	
		if(state == STATE_LEADER_ELECTION){
	
			update_data_structures(msg);
		
			struct msg_prepare_ok *pok = deserialize_prepare_ok(msg);
			if(view_prepared_ready(pok->view)){
				DEBUGPRINT("Prepared Ready; Shifting to Reg Leader\n", NULL);
				shift_to_reg_leader();
			}
		}
	
	}else if(msg_type == PROP){
	
		update_data_structures(msg);
		struct msg_proposal* prop = deserialize_proposal(msg);
		struct msg_accept* acc = create_accept(my_id,prop->view,prop->seq);
		sync();
		send_to_all(serialize_accept(acc));
		update_data_structures(serialize_accept(acc));
		DEBUGPRINT("Proposal [VIEW=%d,SEQ=%d,UPDATE=%s] ACCEPTED\n",prop->view,prop->seq,prop->update);
	
	}else if(msg_type == ACCEPT){
		//DEBUGPRINT("In Accept\n",NULL);
		update_data_structures(msg);
		struct msg_accept *acc = deserialize_accept(msg);
		struct global_history *ghist = get_history_entry(acc->seq);
		//DEBUGPRINT("Before GOU Check\n",NULL);
		
		int x = globally_ordered_ready(acc->seq);
		
		//DEBUGPRINT("GOUReady is %d\n",x);
		
		if(globally_ordered_ready(acc->seq) && !in_global_history(acc->seq)){
			
			struct msg_globally_ordered_update *gou = 								create_gou(my_id,acc->seq,ghist->prop->update);
			//DEBUGPRINT("GOU READY\n",NULL);
			
			update_data_structures(serialize_gou(gou));
			
			//DEBUGPRINT("Updated GOU Data Structures\n",NULL);
			advance_aru();
			//DEBUGPRINT("Advanced ARU\n",NULL);
		}
		
		//DEBUGPRINT("Done With Accept\n",NULL);
	
	}else if(msg_type == CU){
		//DEBUGPRINT("Reached Client Update\n",NULL);
		struct msg_client_update *tmp_cu;
		tmp_cu = deserialize_client_update(msg);
		
		if(my_id == tmp_cu->server_id)
			add_client(tmp_cu->client_id, socket);
		
		if(client_update_handler(tmp_cu)) {
			//send to leader
			//DEBUGPRINT("Forwarding Client Update to Leader\n",NULL);
			send_to_one(msg,coordinator_hostname,1);	
		}
		
		//DEBUGPRINT("!!!!! ***** CLIENT UPDATE FINISHED\n",NULL);
	
	}else if(msg_type == HOST_ID) {
		char *tmp_msg;
		tmp_msg = host_id_response();
		write(socket,tmp_msg,strlen(tmp_msg));
		/* Remove Disconnected Socket from Select */
		FD_CLR(socket,&master);
		FD_CLR(socket,&read_fds);
		close(socket);
		//DEBUGPRINT("sent host id request\n", NULL);
	}else{
		DEBUGPRINT("Unrecognized Message Type\n", NULL);
	}
	//DEBUGPRINT("Returning to Exchange Messages\n",NULL);
}


/* Clean up */
void close_sockets(){

	struct host_entry *tmp;
	struct list_head *pos;
	
	list_for_each(pos, &_hosts.list) {
		tmp = list_entry(pos, struct host_entry, list);
		close(tmp->socket);
	}
}

/* Send a message to all */
void send_to_all(char* msg){

	DEBUGPRINT("Broadcast Message: %s\n",msg);

	/* Send Message to All Participants */
	struct host_entry *tmp;
	struct list_head *pos;
	
	list_for_each(pos, &_hosts.list) {
		tmp = list_entry(pos, struct host_entry, list);
		
		if(tmp == NULL){
			continue;
		}

		send_to_one(msg,tmp->host_name,0);
	}
}

/* Send msg to host */
int send_to_one(char* msg, char* host, int p){

	/* Find Socket for Destination Host */
	int socket = get_host_socket(host);

	/* Ensure Socket Available */		
	if(!strcmp(local_hostname,host)){
		return -1;
	}else if(socket == -1){
		DEBUGPRINT("Could not find socket for host %s\n",host);
		return -1;
	}

	/* Store Size of Message */
	char l[6];
	my_itoa(strlen(msg), l);
	l[5] = '\0';

	/* Tell Receiver Size of Message */
	int b = send(socket, l, 6,MSG_NOSIGNAL);
	
	if(b == 1){
		return -1;
	}

	int nbytes = send(socket,msg,strlen(msg),MSG_NOSIGNAL);
	
	if(nbytes == 1){
		return -1;
	}

	if(nbytes <= 0){
		DEBUGPRINT("Error: Could not send %s to %s\n",msg,host);
		return -1;
	}
	
	if(p == 1)
		DEBUGPRINT("Sent %s to %s\n",msg,host);

	return 1;
}

int send_to_socket(int socket, char* msg){

	//DEBUGPRINT("SEND TO SOCKET: Sending %s to %d\n",msg,socket);

	/* Store Size of Message */
	char l[6];
	my_itoa(strlen(msg), l);
	l[5] = '\0';

	/* Tell Receiver Size of Message */
	int b = send(socket, l, 6,MSG_NOSIGNAL);
	
	if(b == 1){
		return -1;
	}

	int nbytes = send(socket,msg,strlen(msg),MSG_NOSIGNAL);
	
	if(nbytes == 1){
		return -1;
	}

	if(nbytes <= 0){
		DEBUGPRINT("Errno is: %d\n",errno);
		//DEBUGPRINT("!*!*!*!*!*!*!*!*!*!*!*!*!   Error: Could not send %s to %d\n",msg,socket);
		return -1;
	}
	
	DEBUGPRINT("Sent %s to %d\n",msg,socket);
	/* Remove Disconnected Socket from Select */
	FD_CLR(socket,&master);
	FD_CLR(socket,&read_fds);
	close(socket);

	return 1;
}

/* Returns the hostname of an IP address */
char* get_hostname(const char* ipaddr){
	struct in_addr ipv4addr;
	struct hostent *he;

	inet_pton(AF_INET, ipaddr, 	&ipv4addr);
	he = gethostbyaddr(&ipv4addr, sizeof ipv4addr, AF_INET);

	if(he != NULL){
		char* ptr = malloc(strlen(he->h_name)*sizeof(char)+1);
		strncpy(ptr,he->h_name,strlen(he->h_name));
		ptr[strlen(he->h_name)] = '\0';
		return ptr;
	}else{
		DEBUGPRINT("Error: Could not store server hostname\n", NULL);
		return NULL;
	}
}

/* Convert an Integer to a Character String */
int my_itoa(int val, char* buf){

	const unsigned int radix = 10;

	char* p;
	unsigned int a;
	int len;
	char* b;
	char temp;
	unsigned int u;

	p = buf;

	if (val < 0)
	{
		*p++ = '-';
		val = 0 - val;
	}

	u = (unsigned int)val;
	b = p;

	do{
		a = u % radix;
		u /= radix;

		*p++ = a + '0';

	} while (u > 0);

	len = (int)(p - buf);
	*p-- = 0;

	do{
		temp = *p;
		*p = *b;
		*b = temp;
		--p;
		++b;

	} while (b < p);

	return len;
}
